package mylab.starters.mq.impl.redis;


import mylab.starters.cache.support.redis.RedisNotice;
import mylab.starters.mq.api.MessageListener;
import mylab.starters.mq.api.Properties;
import mylab.starters.mq.impl.AbstractConsumer;
import mylab.utils.spring.SpringUtil;
import org.springframework.data.redis.connection.Message;


public class RedisConsumer extends AbstractConsumer {

    RedisNotice redisNotice;
    RedisListener listener;

    public RedisConsumer(Properties properties, String groupId, String clientId) {
        super(properties, groupId, clientId);
        redisNotice = SpringUtil.getBean(RedisNotice.class);
    }

    @Override
    protected void onSubscribe(String topic, MessageListener messageListener, String... tags) throws Exception {
        listener = new RedisListener(topic, messageListener);
        redisNotice.subscribe(listener, topic);
    }

    @Override
    protected void onUnsubscribe() throws Exception {
        redisNotice.unsubscribe(listener, topic);
    }

    public static class RedisListener implements org.springframework.data.redis.connection.MessageListener {
        String topic;
        MessageListener listener;

        public RedisListener(String topic, MessageListener listener) {
            this.topic = topic;
            this.listener = listener;
        }

        @Override
        public void onMessage(Message message, byte[] pattern) {
            String content = new String(message.getBody());

            mylab.starters.mq.api.Message msg = new mylab.starters.mq.api.Message(topic, content);
            listener.onSuccess(msg);
        }
    }
}
